package org.springframework.boot.netty.handler;

import com.google.protobuf.MessageLite;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.ReferenceCountUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.netty.codec.ProtoBufCodecBean;
import org.springframework.boot.netty.codec.proto.SpeakMessage;
import org.springframework.boot.netty.converter.MessageConverter;
import org.springframework.boot.netty.converter.ProtobufMessage2JsonConverter;
import org.springframework.boot.netty.exception.ConvertException;
import org.springframework.boot.netty.handler.common.AbstractChannelInboundHandlerAdapter;
import org.springframework.boot.netty.listener.Message;
import org.springframework.boot.netty.sync.MessageBlockQueue;
import org.springframework.context.ApplicationContext;

/**
 * Author: huoxingzhi
 * Date: 2020/12/14
 * Email: hxz_798561819@163.com
 */
public class DefaultProtoBufServerHandler extends AbstractChannelInboundHandlerAdapter {


    private final static Logger logger = LoggerFactory.getLogger(DefaultProtoBufServerHandler.class);

    private MessageConverter messageConverter;

    public DefaultProtoBufServerHandler() {
    }


    public DefaultProtoBufServerHandler(ApplicationContext applicationContext) {
        this.applicationContext = applicationContext;
        for (MessageConverter messageConverter : this.messageConverters) {
            if(messageConverter.fromSupport(MessageLite.class,String.class)){
                this.messageConverter = messageConverter;
                break;
            }
        }
    }
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 获取客户端传输过来的消息
        if(!ProtoBufCodecBean.class.isAssignableFrom(msg.getClass())){
            throw new ConvertException("DefaultProtoBufServerHandler.clas-->channelRead-->"+msg.getClass()+"can not convert com.google.protobuf.Message.class");
        }

        ProtoBufCodecBean protoBufCodecBean = (ProtoBufCodecBean) msg;
        MessageLite message = protoBufCodecBean.getMessage();

        if(!SpeakMessage.class.isAssignableFrom(message.getClass())){
            String jsonContent = messageConverter.fromMessage(message,String.class);
            String accessToken = protoBufCodecBean.getHeaders().getOrDefault("access_token", "");
            Message<String> serviceMsg = new Message<>();
            serviceMsg.setContent(jsonContent);
            serviceMsg.setAccessToken(accessToken);
            super.channelRead(ctx,serviceMsg);
            // 释放内存
            ReferenceCountUtil.release(msg);
        }

        //心跳消息，交给其他handler
        ctx.fireChannelRead(msg);

    }

}
